﻿/*************************************************************************************
 *
 * 文 件 名:   Worker
 * 描    述:   数据前置接收
 * 
 * 版    本：  V1.0
 * 创 建 者：  zcb-lms 
 * 创建时间：  2021/9/8 9:05:44
 * ======================================
 * 历史更新记录
 * 版本：V1.0       修改时间：2021-09-09    修改人：杨世旺
 * 修改内容：添加kafka生产者池ProducerPool
 * ======================================
*************************************************************************************/

using CommonModel.Util;
using GXRTBTC.DataStorageService.Model;
using GXRTBTC.PublicClassLibrary;
using GXRTBTC.PublicClassLibrary.Field;
using GXRTBTC.PublicClassLibrary.Interface;
using GXRTBTC.PublicClassLibrary.Model;
using GXRTBTC.PublicClassLibrary.ParseImpl;
using GXRTBTC.PublicClassLibrary.Redis;
using GXRTBTC.PublicClassLibrary.Util;
using HPSocket;
using HPSocket.Udp;
using Microsoft.Extensions.Caching.StackExchangeRedis;
using Microsoft.Extensions.Configuration;
using Microsoft.Extensions.Hosting;
using Microsoft.Extensions.Logging;
using Newtonsoft.Json;
using RedisHelp;
using StackExchange.Redis;
using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.Data;
using System.Diagnostics;
using System.Linq;
using System.Threading;
using System.Threading.Tasks;

namespace GXRTBTC.DataReceiveService
{
    /// <summary>
    /// 后台服务
    /// </summary>
    public class Worker : BackgroundService
    {
        #region 参数
        /// <summary>
        /// 当前节点名称
        /// 组成：节点所在IP+容器ID
        /// </summary>
        public string nodeName { get; set; } = TransForm.GetHostIPOrName() + "_" + Environment.MachineName;
        /// <summary>
        /// 心跳节点名称，保证唯一
        /// </summary>
        private string keySvr { get; set; } = "DataReceiveService";
        /// <summary>
        /// 当前最新的心跳包
        /// </summary>
        private Dictionary<string, DateTime> headKeys { get; set; } = new();
        /// <summary>
        /// 日志
        /// </summary>
        private readonly ILogger mLogger;
        /// <summary>
        /// udp监听ipv4
        /// </summary>
        private readonly IUdpNode mUdpNodeIpv4 = new UdpNode();
        /// <summary>
        /// udp监听ipv6
        /// </summary>
        private readonly IUdpNode mUdpNodeIpv6 = new UdpNode();
        /// <summary>
        /// 配置
        /// </summary>
        private readonly IConfiguration mConfiguration;
        private string redisConnString { get; set; } = "";
        /// <summary>
        /// StackExchange.Redis
        /// 初始化、下发、心跳用
        /// </summary>
        private RedisHelper redisInit { get; set; }
        /// <summary>
        /// 写台站、设备在离线用
        /// </summary>
        private RedisHelper redisStatus { get; set; }
        private RedisHelper redisCache { get; set; }

        /// <summary>
        /// 当前节点状态统计信息
        /// </summary>
        private Dictionary<string, Object> nodeStatusInfo { get; set; } = new Dictionary<string, object>() {
            { "Datetime", TransForm.GetFormatDatetime() },
            { "GetSum", 0 },
            { "GetRate", 0 },
            { "SkipNum", 0 },
            { "RedisWaiting",0 },
            { "RedisCache",0 },
            { "nodeNo",0 }
        };

        /// <summary>
        /// 副本数，默认只有自身
        /// </summary>
        private int nodeNum { get; set; } = 1;
        private int nodeSum { get; set; } = 8;
        /// <summary>
        /// 当前副本编号
        /// </summary>
        private int nodeNo { get; set; } = 0;

        /// <summary>
        /// 接收、发送数量差，用于判断是否增加线程池数量
        /// </summary>
        private long gapNums { get; set; } = 0;
        /// <summary>
        /// 心跳包数据
        /// </summary>
        private long HeartbeatNum { get; set; } = 0;
        /// <summary>
        /// 锁对象，用于线程间同步gapNums
        /// </summary>
        private Object syncLock { get; set; } = new object();
        /// <summary>
        /// 跳过不处理的数量
        /// </summary>
        private long skipNum { get; set; } = 0;
        /// <summary>
        /// 锁对象
        /// </summary>
        private Object skipLock { get; set; } = new object();
        /// <summary>
        /// 接收数据总量
        /// </summary>
        private long numsRcv { get; set; } = 0;
        /// <summary>
        /// 数据是否特殊处理
        /// </summary>
        private bool SpecialFlow { get; set; } = true;
        /// <summary>
        /// 最小线程池数量
        /// </summary>
        private int minPoolNum { get; set; } = 20;
        /// <summary>
        /// 最大线程池数量
        /// </summary>
        private int maxPoolNum { get; set; } = 100;
        /// <summary>
        /// 采集数据最新地址端口记录
        /// </summary>
        private ConcurrentDictionary<string, List<DateTime>> rcvDataRecords { get; set; } = new();
        private ConcurrentDictionary<string, int> rcvErrDevs { get; set; } = new();

        /// <summary>
        /// targetStation：台站ip=》台站编号
        /// targetPositionId：台站ip:port=》工位号
        /// </summary>
        private Dictionary<string, string> targetStation { get; set; } = new Dictionary<string, string>();
        private Dictionary<string, string> targetPositionId { get; set; } = new Dictionary<string, string>();
        private Dictionary<string, int> stationsAddrToLevels { get; set; } = new();
        private Dictionary<string, bool> saveStationLevels { get; set; } = new Dictionary<string, bool>() { { "1", true }, { "2", true }, { "3", true }, { "4", true }, { "5", true } };
        private List<string> positionIds { get; set; } = new List<string>();
        private Dictionary<string, string> devSaveTimePre { get; set; } = new();

        /// <summary>
        /// 台站信息缓存
        /// 台站编号为key
        /// </summary>
        private Dictionary<string, Dictionary<string, string>> stationsInfo { get; set; } = new Dictionary<string, Dictionary<string, string>>();
        private string stationIdGlobal { get; set; } = "";
        /// <summary>
        /// 台站信息缓存
        /// 台站ip地址为key
        /// </summary>
        private Dictionary<string, Dictionary<string, string>> stationAddrs { get; set; } = new Dictionary<string, Dictionary<string, string>>();
        /// <summary>
        /// 六路调谐器IP+端口，匹配都台站ID
        /// </summary>
        private Dictionary<string, Dictionary<string, Dictionary<string, string>>> tunerMatchInfo { get; set; } = new();
        /// <summary>
        /// dtmb远场IP-端口匹配
        /// </summary>
        private Dictionary<string, Dictionary<string, Dictionary<string, Dictionary<string, string>>>> dtmbMatchInfo { get; set; } = new();
        /// <summary>
        /// 工位号匹配信息
        /// </summary>
        private Dictionary<string, Dictionary<string, Dictionary<string, Dictionary<string, string>>>> matchInfos { get; set; } = new Dictionary<string, Dictionary<string, Dictionary<string, Dictionary<string, string>>>>();
        /// <summary>
        /// 工位号和频率对应
        /// </summary>
        public Dictionary<string, Dictionary<string, Dictionary<string, string>>> positionIdToFrqs { get; set; } = new();

        /// <summary>
        /// 设备信息
        /// </summary>
        private Dictionary<string, DeviceInfo> deviceInfos { get; set; } = new Dictionary<string, DeviceInfo>();
        /// <summary>
        /// 数据缓存
        /// </summary>
        public BlockingCollection<PacketInfo> dataCacheGet { get; set; } = null; //Redis做消息队列
        public BlockingCollection<PacketInfo> dataCacheSet { get; set; } = null; //Redis做消息队列
        /// <summary>
        /// ip:port缓存
        /// 用于更新台站、设备在线情况
        /// </summary>
        public BlockingCollection<string> targetCache { get; set; } = null; //Redis做消息队列
        /// <summary>
        /// IP:PORT端口最新时间记录
        /// </summary>
        private Dictionary<string, DateTime> targetRec { get; set; } = new();
        /// <summary>
        /// 是否接收到数据
        /// </summary>
        private bool isRcvData { get; set; } = false;
        private DateTime dtBegin { get; set; } = DateTime.Now;
        /// <summary>
        /// 最新接收时间
        /// </summary>
        private DateTime dtNewestTime { get; set; } = DateTime.Now;

        /// <summary>
        /// 配置集
        /// </summary>
        private Dictionary<string, string> paramInfo { get; set; } = new();

        /// <summary>
        /// 转换解析类
        /// </summary>
        private Dictionary<string, IConvertParse> ConvertParseImpls { get; set; }
        /// <summary>
        /// 配置信息
        /// </summary>
        private Dictionary<string, SchemaDocument> schemaDict { get; set; } = new Dictionary<string, SchemaDocument>();
        /// <summary>
        /// 每个类型对应一个表信息
        /// </summary>
        public Dictionary<int, TableInfo> tableInfos { get; set; } = new Dictionary<int, TableInfo>();
        /// <summary>
        /// //30W模拟调频发射机，50W自研数字电视发射机和水冷系统
        /// </summary>
        private ConvertParse_30Wor50W_Digit50W_Impl convertParse_30Wor50W_Digit50W_Impl { get; set; }
        /// <summary>
        /// 00/1000W 数字电视发射机主动上报
        /// </summary>
        private ConvertParse_ByteParseToHeadDigitTransmitter300_Impl convertParse_ByteParseToHeadDigitTransmitter300_Impl { get; set; }
        /// <summary>
        /// 远程监控通信协议规范（主控协议）
        /// </summary>
        private ConvertParse_General_GXGD_Impl convertParse_General_GXGD_Impl { get; set; }
        /// <summary>
        /// 四级环境采集器，四级电源复位器
        /// </summary>
        private ConvertParse_Four_Enviroment_Power_NoHeader_Impl convertParse_Four_Enviroment_Power_NoHeader_Impl { get; set; }
        /// <summary>
        /// snmp
        /// </summary>
        private ConvertParse_SNMP_Impl convertParse_SNMP_Impl { get; set; }
        #endregion

        public Worker(ILogger<Worker> logger, IConfiguration configuration)
        {
            this.mLogger = logger;
            this.mConfiguration = configuration;

            //给通用日志赋值
            CommonUtil.mLogger = logger;

            //初始化
            Init();
        }
        #region 初始化
        /// <summary>
        /// 初始化
        /// </summary>
        private void Init()
        {
            //线程池配置
            ThreadPool.SetMinThreads(minPoolNum, minPoolNum);
            ThreadPool.SetMaxThreads(maxPoolNum, maxPoolNum);

            try
            {
                //redis初始化
#if DEBUG
                redisConnString = RsaEncrypt.DecryptFree(mConfiguration["RedisConnStringDebug"]);
#else
                redisConnString = RsaEncrypt.DecryptFree(mConfiguration["RedisConnString"]);
#endif

                //rdis实例，全局用
                redisInit = new RedisHelper(redisConnString, mLogger);
                redisStatus = new RedisHelper(redisConnString, mLogger);
                redisCache = new RedisHelper(redisConnString, mLogger);
            }
            catch { }

            //从缓存读取设置
            paramInfo = redisInit.HashGetAllDict<string>("paramInfo");
            //缓存信息初始化
            InitCacheInfo();

            //数据缓存
            dataCacheGet = new BlockingCollection<PacketInfo>();
            dataCacheSet = new BlockingCollection<PacketInfo>();
            //台站、设备在线情况缓存
            targetCache = new BlockingCollection<string>();

            string rcvStationLevel = paramInfo.GetValueOrDefault("RcvStationLevels", "1,2,3,4,5");
            string rcvNodeSum = paramInfo.GetValueOrDefault("RcvNodeSum", "8");
            nodeSum = (!"".Equals(rcvNodeSum) && TransForm.IsInt(rcvNodeSum)) ? int.Parse(rcvNodeSum) : nodeSum;

            //配置参数，保存哪个台站级别的数据
            //if (mConfiguration["SaveStationLevel"] != null && !"".Equals(mConfiguration["SaveStationLevel"]))
            if (rcvStationLevel != null && !"".Equals(rcvStationLevel))
            {
                //string[] tmpSaveStationLevel = mConfiguration["SaveStationLevel"].Split(',');
                string[] tmpSaveStationLevel = rcvStationLevel.Split(',');
                if (tmpSaveStationLevel.Length > 0)
                {
                    Dictionary<string, bool> tmpSaveStationLevels = new Dictionary<string, bool>();
                    foreach (string level in tmpSaveStationLevel)
                    {
                        if (!tmpSaveStationLevels.ContainsKey(level))
                        {
                            tmpSaveStationLevels.TryAdd(level, true);
                        }
                    }
                    saveStationLevels = tmpSaveStationLevels;
                }
            }

            //数据处理解析类初始化
            convertParse_30Wor50W_Digit50W_Impl = new ConvertParse_30Wor50W_Digit50W_Impl() { schemaDict = schemaDict };
            convertParse_ByteParseToHeadDigitTransmitter300_Impl = new ConvertParse_ByteParseToHeadDigitTransmitter300_Impl() { schemaDict = schemaDict };
            convertParse_General_GXGD_Impl = new ConvertParse_General_GXGD_Impl() { schemaDict = schemaDict };
            convertParse_Four_Enviroment_Power_NoHeader_Impl = new ConvertParse_Four_Enviroment_Power_NoHeader_Impl() { schemaDict = schemaDict };
            convertParse_SNMP_Impl = new ConvertParse_SNMP_Impl() { schemaDict = schemaDict.Where(d => d.Key.ToUpper().Equals("0X0708_SNMP_GETBASICMSG")).ToDictionary(d => d.Key, d => d.Value) };

            ConvertParseImpls = new Dictionary<string, IConvertParse>
            {
                { convertParse_30Wor50W_Digit50W_Impl.mVersion, convertParse_30Wor50W_Digit50W_Impl },
                { convertParse_ByteParseToHeadDigitTransmitter300_Impl.mVersion, convertParse_ByteParseToHeadDigitTransmitter300_Impl },
                { convertParse_General_GXGD_Impl.mVersion, convertParse_General_GXGD_Impl },
                { convertParse_Four_Enviroment_Power_NoHeader_Impl.mVersion, convertParse_Four_Enviroment_Power_NoHeader_Impl },
                { convertParse_SNMP_Impl.mVersion, convertParse_SNMP_Impl }
            };
        }

        /// <summary>
        /// redis缓存信息初始化
        /// </summary>
        private void InitCacheInfo()
        {
            try
            {
                Dictionary<string, string> tmpTargetStation = redisInit.HashGetAllDict<string>("ipaddr_station");
                Dictionary<string, string> tmpTargetPositionId = redisInit.HashGetAllDict<string>("target_position");
                if (tmpTargetStation != null)
                {
                    lock (targetStation)
                    {
                        targetStation = tmpTargetStation;
                    }
                }
                if (tmpTargetPositionId != null)
                {
                    lock (targetPositionId)
                    {
                        targetPositionId = tmpTargetPositionId;
                    }
                }

                //台站信息缓存
                //台站编号地址为key
                //台站信息缓存
                //台站编号地址为key
                Dictionary<string, Dictionary<string, string>> tmpStationsInfo = redisInit.StringGet<Dictionary<string, Dictionary<string, string>>>("stations");
                stationsInfo = (tmpStationsInfo != null && tmpStationsInfo.Count > 0) ? tmpStationsInfo : stationsInfo;
                if (tmpStationsInfo != null && tmpStationsInfo.Count > 0)
                { 
                    stationIdGlobal = tmpStationsInfo.Keys.FirstOrDefault();
                    foreach (string stationId in tmpStationsInfo.Keys)
                    {
                        string ipAddr = tmpStationsInfo[stationId]["ip_addr"];
                        string stationLevel = tmpStationsInfo[stationId]["station_level"];
                        if (!stationsAddrToLevels.ContainsKey(ipAddr))
                        {
                            lock (stationsAddrToLevels)
                            {
                                stationsAddrToLevels.TryAdd(ipAddr, int.Parse(stationLevel));
                            }
                        }
                    }
                }

                //redis缓存的所有工位号keys
                List<string> cachePositionIds = redisInit.StringGet<List<string>>("positionIds");
                List<string> tmpPositionIds = new List<string>();
                if (cachePositionIds != null && cachePositionIds.Count > 0)
                {
                    foreach (string pid in cachePositionIds)
                    {
                        string stationId = pid.Split("_")[0];
                        if (tmpStationsInfo.ContainsKey(stationId))
                        {
                            int stationLevel = int.Parse(tmpStationsInfo[stationId]["station_level"]);
                            if (stationLevel < 4)
                            {
                                tmpPositionIds.Add(pid);
                            }
                        }
                    }

                    positionIds = tmpPositionIds;
                }

                //配置信息
                Dictionary<string, SchemaDocument> tmpSchemaDict = redisInit.HashGetAllDict<SchemaDocument>("pre_analysis_config");
                if (tmpSchemaDict != null && tmpSchemaDict.Count > 0)
                {
                    lock (schemaDict)
                    {
                        schemaDict = tmpSchemaDict;
                    }

                    if (ConvertParseImpls != null)
                    {
                        foreach (IConvertParse convertParse in ConvertParseImpls.Values)
                        {
                            lock (convertParse.schemaDict)
                            {
                                convertParse.schemaDict = schemaDict;
                            }
                        }
                    }
                }

                //台站信息缓存
                //台站ip地址为key
                Dictionary<string, Dictionary<string, string>> tmpStationAddrs = redisInit.StringGet<Dictionary<string, Dictionary<string, string>>>("stationAddrs");
                stationAddrs = (tmpStationAddrs != null && tmpStationAddrs.Count > 0) ? tmpStationAddrs : stationAddrs;

                Dictionary<string, Dictionary<string, Dictionary<string, string>>> tempTunerMatchInfo = redisInit.StringGet<Dictionary<string, Dictionary<string, Dictionary<string, string>>>>("tunerMatchInfo");
                tunerMatchInfo = (tempTunerMatchInfo != null && tempTunerMatchInfo.Count > 0) ? tempTunerMatchInfo : tunerMatchInfo;

                Dictionary<string, Dictionary<string, Dictionary<string, Dictionary<string, string>>>> tempDtmbMatchInfo = redisInit.StringGet<Dictionary<string, Dictionary<string, Dictionary<string, Dictionary<string, string>>>>>("dtmbMatchInfo");
                dtmbMatchInfo = (tempDtmbMatchInfo != null && tempDtmbMatchInfo.Count > 0) ? tempDtmbMatchInfo : dtmbMatchInfo;

                //工位号匹配生成信息
                //一级key为台站基本，二级key为端口
                Dictionary<string, Dictionary<string, Dictionary<string, Dictionary<string, string>>>> tmpMatchInfos = redisInit.StringGet<Dictionary<string, Dictionary<string, Dictionary<string, Dictionary<string, string>>>>>("workMatchs");
                //matchInfos = (tmpMatchInfos != null && tmpMatchInfos.Count > 0) ? tmpMatchInfos : matchInfos;
                if (tmpMatchInfos != null && tmpMatchInfos.Count > 0)
                {
                    foreach (string key in tmpMatchInfos.Keys)
                    {
                        if (tmpMatchInfos[key] != null && tmpMatchInfos[key].Count > 0)
                        {
                            if (!matchInfos.ContainsKey(key))
                            {
                                matchInfos.TryAdd(key, tmpMatchInfos[key]);
                            }
                            else
                            {
                                matchInfos[key] = tmpMatchInfos[key];
                            }
                        }
                    }
                }

                //工位号和频率对应
                Dictionary<string, Dictionary<string, Dictionary<string, string>>> tmpPositionIdToFrqs = redisInit.HashGetAllDict<Dictionary<string, Dictionary<string, string>>>("positionIdToFrq");
                if (tmpPositionIdToFrqs != null && tmpPositionIdToFrqs.Count > 0)
                {
                    lock (tmpPositionIdToFrqs)
                    {
                        positionIdToFrqs = tmpPositionIdToFrqs;
                    }
                }

                //deviceInfos
                var tmpDeviceInfos = redisInit.HashGetAllDict<DeviceInfo>("station_deviceInfos");
                List<string> pids = deviceInfos.Keys.ToList();
                foreach (string positionId in pids)
                {
                    if (!tmpDeviceInfos.ContainsKey(positionId) || deviceInfos[positionId].new_time > tmpDeviceInfos[positionId].new_time)
                    {
                        redisInit.HashSet("station_deviceInfos", positionId, deviceInfos[positionId]);
                    }
                }

                Dictionary<string, TableInfo> tmpTableInfos = redisInit.HashGetAllDict<TableInfo>("pre_save_tableInfo");
                if (tmpTableInfos != null && tmpTableInfos.Count > 0)
                {
                    lock (tableInfos)
                    {
                        tableInfos = tmpTableInfos.ToDictionary(d => int.Parse(d.Key), d => d.Value);
                    }
                }
            }
            catch (Exception e)
            {
                mLogger.LogInformation("更新获取台站信息缓存失败，" + e);
            }
        }
        #endregion

        #region 监听
        /// <summary>
        /// 开启UdpNode监听
        /// </summary>
        private void UdpNodeInit()
        {
            //启动IPV4监听
            mUdpNodeIpv4.Port = 8000;
            mUdpNodeIpv4.FreeBufferPoolSize = 5120;
            mUdpNodeIpv4.FreeBufferPoolHold = 5120;
            mUdpNodeIpv4.OnReceive += UdpNode_OnReceive; ;
            mUdpNodeIpv4.OnError += UdpNode_OnError;
            mUdpNodeIpv4.OnShutdown += UdpNode_OnShutdown;
            mUdpNodeIpv4.OnPrepareListen += UdpNode_OnPrepareListen;
            mUdpNodeIpv4.Start();

            //启动IPV6监听
            mUdpNodeIpv6.Address = "::";
            mUdpNodeIpv6.FreeBufferPoolSize = 5120;
            mUdpNodeIpv6.FreeBufferPoolHold = 5120;
            mUdpNodeIpv6.Port = 8000;
            mUdpNodeIpv6.OnReceive += UdpNode_OnReceive; ;
            mUdpNodeIpv6.OnError += UdpNode_OnError;
            mUdpNodeIpv6.OnShutdown += UdpNode_OnShutdown;
            mUdpNodeIpv6.OnPrepareListen += UdpNode_OnPrepareListen;
            mUdpNodeIpv6.Start();
        }

        private HandleResult UdpNode_OnError(IUdpNode sender, SocketOperation socketOperation, int errorCode, string remoteAddress, ushort remotePort, byte[] data)
        {
            mLogger.LogError("UdpNode出现错误" + errorCode);
            return HandleResult.Ok;
        }

        private HandleResult UdpNode_OnShutdown(IUdpNode sender)
        {
            mLogger.LogError($"UdpNode关闭:{sender.Address}:{sender.Port}");
            return HandleResult.Ok;
        }

        private HandleResult UdpNode_OnPrepareListen(IUdpNode sender, IntPtr soListen)
        {
            mLogger.LogInformation($"启动了UDP监听:{sender.Address}:{sender.Port}");
            return HandleResult.Ok;
        }
        #endregion

        #region 收到数据
        /// <summary>
        /// 接收数据
        /// </summary>
        /// <param name="sender"></param>
        /// <param name="remoteAddress"></param>
        /// <param name="remotePort"></param>
        /// <param name="pData"></param>
        /// <returns></returns>
        private HandleResult UdpNode_OnReceive(IUdpNode sender, string remoteAddress, ushort remotePort, byte[] pData)
        {
            string hexData = TransForm.byteToHex(pData);
            try
            {
                //杨世旺测试电脑的IP，直接返回
                if (remoteAddress.Equals("10.170.0.197"))
                {
                    sender.Send(remoteAddress, remotePort, pData, pData.Length);
                }

                //数据长度过滤
                if (pData.Length < 5)
                {
                    return HandleResult.Ok;
                }

                //保存哪些数据
                if (stationsAddrToLevels.ContainsKey(remoteAddress) && !saveStationLevels.ContainsKey(stationsAddrToLevels[remoteAddress].ToString()))
                {
                    return HandleResult.Ok;
                }

                /***************************有个别设备数据上报异常，这里过滤掉部分密集上报的数据，确保系统稳定****************************************************/
                DateTime dtNow = DateTime.Now;
                string dataKey = remoteAddress + ":" + remotePort + "_" + pData.Length + "_" + TransForm.StrToMD5Hash(hexData);
                lock (rcvDataRecords)
                {
                    if (!rcvDataRecords.ContainsKey(dataKey))
                    {
                        rcvDataRecords.TryAdd(dataKey, new List<DateTime>() { dtNow });
                    }
                    else
                    {
                        rcvDataRecords[dataKey].Add(dtNow);
                        //rcvDataRecords[dataKey] = rcvDataRecords[dataKey].Where(l => (dtNow - l).TotalSeconds <= 1).ToList();
                        //改为两秒
                        rcvDataRecords[dataKey] = rcvDataRecords[dataKey].Where(l => (dtNow - l).TotalSeconds <= 2).ToList();
                    }
                }


                //单个设备最近一定时间超过一定数量，目前是一秒钟5条
                //if (rcvDataRecords[dataKey].Count > 5)
                //改为两秒11条
                if (rcvDataRecords[dataKey].Count > 11)
                {
                    string from = remoteAddress + ":" + remotePort;
                    //记录下来，打印到输出日志
                    lock (rcvErrDevs)
                    {
                        if (!rcvErrDevs.ContainsKey(from))
                        {
                            rcvErrDevs.TryAdd(from, 1);
                        }
                        else
                        {
                            rcvErrDevs[from] += 1;
                        }
                    }

                    return HandleResult.Ok;
                }
                /***************************有个别设备数据上报异常，这里过滤掉部分密集上报的数据，确保系统稳定****************************************************/
            }
            catch (Exception e)
            {
                mLogger.LogError("判断异常，" + e);
                return HandleResult.Ok;
            }

            //如果堆积超过阈值（比如2千）条数据，就跳过
            if (gapNums > 2000)
            {
                lock (skipLock)
                {
                    skipNum++;
                }
                return HandleResult.Ok;
            }

            //组装
            PacketInfo packetInfo = new PacketInfo
            {
                //reviceDatas = TransForm.byteToHex(pData),
                reviceDatas = hexData,
                ip = remoteAddress,
                port = remotePort,
                receivingTime = DateTime.Now,
                receiveNode = nodeName + "_" + (nodeNo + 1).ToString(),
                receiveNo = ((nodeNo % nodeSum) + 1).ToString()
            };

            dtNewestTime = DateTime.Now;

            //记录来源
            string key = packetInfo.ip + ":" + packetInfo.port;
            lock (targetRec)
            {
                if (!targetRec.ContainsKey(key))
                {
                    targetRec.TryAdd(key, DateTime.Now);
                }
                else
                {
                    targetRec[key] = DateTime.Now;
                }
            }

            //收入线程队列
            lock (dataCacheGet)
            {
                dataCacheGet.Add(packetInfo);
            }
            lock (targetCache)
            {
                targetCache.Add(remoteAddress + ":" + remotePort + "," + JsonConvert.SerializeObject(packetInfo.receivingTime));
            }
            lock (syncLock)
            {
                numsRcv++;
                if (!isRcvData)
                {
                    isRcvData = true;
                    mLogger.LogInformation("接收到数据，开始处理");
                }
            }

            return HandleResult.Ok;
        }
        #endregion

        #region 数据分组、处理
        /// <summary>
        /// 数据分组
        /// </summary>
        private void DealData()
        {
            Thread.Sleep(new Random().Next(500, 5000));
            while (true)
            {
                try
                {
                    if (dataCacheGet.Count > 0)
                    {
                        while (dataCacheGet.Count > 0)
                        {
                            PacketInfo packetInfo;
                            bool packError = true;
                            lock (dataCacheGet)
                            {
                                packetInfo = dataCacheGet.Take();
                            }

                            //超时时间判断，超过配置分钟数则跳过不提交
                            TimeSpan ts = DateTime.Now - packetInfo.receivingTime;

                            string addrKey = packetInfo.ip;    //当前台站根据IP地址记录最新在线时间
                            string from = addrKey + ":" + packetInfo.port.ToString();   //记录来源


                            //0-普通数据，需要配置解析配置数据
                            //1-json数据，收到的就是json数据
                            if ("0".Equals(packetInfo.dataType))
                            {
                                //将头部识别提前
                                //识别帧头
                                IConvertParse convertParse = MachingParse(TransForm.HexTobyte(packetInfo.reviceDatas)); //获取转换实现类
                                if (convertParse != null)
                                {
                                    //获取头信息
                                    packetInfo.header = convertParse.GetHeaderData(packetInfo);
                                }

                                //六路调谐器比较特殊，先判断
                                //是否六路调谐器
                                bool tuner = false;
                                //是否远场dtmb
                                bool ycdtmb = false;
                                //六路调谐器工位号编码
                                string tunerNo = "1";
                                string ycdtmbNo = "2";
                                string tunerDeviceType = "0x0901";
                                string ycdtmbDeviceType = "0x0902";

                                //台站IP匹配
                                //if (stationAddrs.ContainsKey(addrKey))
                                //{
                                //    //得到台站信息
                                //    packetInfo.stationCode = stationAddrs[addrKey]["station_id"];
                                //    packetInfo.stationLevel = stationAddrs[addrKey]["station_level"];
                                //}
                                packetInfo.stationCode = stationIdGlobal;
                                packetInfo.stationLevel = stationsInfo[packetInfo.stationCode]["station_level"];

                                //不放在else里面，因为发现1~3台站的远场收测，可能会放在4级台站返回
                                if (tunerMatchInfo.ContainsKey(addrKey) && tunerMatchInfo[addrKey].ContainsKey(packetInfo.port.ToString())
                                   && stationsInfo.ContainsKey(tunerMatchInfo[addrKey][packetInfo.port.ToString()]["station_id"]))
                                {
                                    //得到台站信息
                                    tuner = true;
                                    tunerNo = tunerMatchInfo[addrKey][packetInfo.port.ToString()]["num"];
                                    tunerDeviceType = tunerMatchInfo[addrKey][packetInfo.port.ToString()]["device_type"];
                                    string stationCode = tunerMatchInfo[addrKey][packetInfo.port.ToString()]["station_id"];
                                    if (!"".Equals(packetInfo.stationCode) && !packetInfo.stationCode.Equals(stationCode))
                                    {
                                        packetInfo.tipMsg += "【" + stationCode + "】的数据来自于台站【" + packetInfo.stationCode + "】的地址：" + addrKey;
                                    }
                                    packetInfo.stationCode = tunerMatchInfo[addrKey][packetInfo.port.ToString()]["station_id"];
                                    packetInfo.stationLevel = stationsInfo[packetInfo.stationCode]["station_level"];
                                }
                                //20240221 包含IP，且device_type匹配，且只有一个设置，默认匹配
                                else if (tunerMatchInfo.ContainsKey(addrKey) && tunerMatchInfo[addrKey].Count == 1)
                                {
                                    foreach (string tmpPort in tunerMatchInfo[addrKey].Keys)
                                    {
                                        if (tunerMatchInfo[addrKey][tmpPort]["device_type"].Equals(packetInfo.header.device)
                                            && tunerMatchInfo[addrKey][tmpPort]["station_id"].Equals(packetInfo.header.stationcode))
                                        {
                                            //得到台站信息
                                            tuner = true;
                                            tunerNo = tunerMatchInfo[addrKey][tmpPort]["num"];
                                            tunerDeviceType = tunerMatchInfo[addrKey][tmpPort]["device_type"];
                                            string stationCode = tunerMatchInfo[addrKey][tmpPort]["station_id"];
                                            packetInfo.tipMsg += "端口" + packetInfo.port + "未匹配" + tmpPort + "，系统默认通过解析，保留该提示";
                                            packetInfo.stationCode = tunerMatchInfo[addrKey][tmpPort]["station_id"];
                                            packetInfo.stationLevel = stationsInfo[packetInfo.stationCode]["station_level"];
                                        }
                                    }
                                }

                                //20230314 如果是远场DTMB接收机的
                                if (dtmbMatchInfo.ContainsKey(addrKey) && dtmbMatchInfo[addrKey]["port"].ContainsKey(packetInfo.port.ToString())
                                   && stationsInfo.ContainsKey(dtmbMatchInfo[addrKey]["port"][packetInfo.port.ToString()]["station_id"]))
                                {
                                    //得到台站信息
                                    ycdtmb = true;
                                    ycdtmbNo = dtmbMatchInfo[addrKey]["port"][packetInfo.port.ToString()]["num"];
                                    //tunerDeviceType = dtmbMatchInfo[addrKey]["port"][packetInfo.port.ToString()]["device_type"];
                                    ycdtmbDeviceType = dtmbMatchInfo[addrKey]["port"][packetInfo.port.ToString()]["device_type"];
                                    string stationCode = dtmbMatchInfo[addrKey]["port"][packetInfo.port.ToString()]["station_id"];
                                    if (!"".Equals(packetInfo.stationCode) && !packetInfo.stationCode.Equals(stationCode))
                                    {
                                        packetInfo.tipMsg += "【" + stationCode + "】的数据来自于台站【" + packetInfo.stationCode + "】的地址：" + addrKey;
                                    }
                                    packetInfo.stationCode = dtmbMatchInfo[addrKey]["port"][packetInfo.port.ToString()]["station_id"];
                                    packetInfo.stationLevel = stationsInfo[packetInfo.stationCode]["station_level"];
                                }
                                //20230314 如果是远场DTMB接收机的
                                else if (dtmbMatchInfo.ContainsKey(addrKey) && dtmbMatchInfo[addrKey]["addr"].ContainsKey(packetInfo.header.devaddr.ToString())
                                   && stationsInfo.ContainsKey(dtmbMatchInfo[addrKey]["addr"][packetInfo.header.devaddr.ToString()]["station_id"]))
                                {
                                    //得到台站信息
                                    ycdtmb = true;
                                    ycdtmbNo = dtmbMatchInfo[addrKey]["addr"][packetInfo.header.devaddr.ToString()]["num"];
                                    //tunerDeviceType = dtmbMatchInfo[addrKey]["addr"][packetInfo.header.devaddr.ToString()]["device_type"];
                                    ycdtmbDeviceType = dtmbMatchInfo[addrKey]["addr"][packetInfo.header.devaddr.ToString()]["device_type"];
                                    string stationCode = dtmbMatchInfo[addrKey]["addr"][packetInfo.header.devaddr.ToString()]["station_id"];
                                    if (!"".Equals(packetInfo.stationCode) && !packetInfo.stationCode.Equals(stationCode))
                                    {
                                        packetInfo.tipMsg += "【" + stationCode + "】的数据来自于台站【" + packetInfo.stationCode + "】的地址：" + addrKey;
                                    }
                                    packetInfo.stationCode = dtmbMatchInfo[addrKey]["addr"][packetInfo.header.devaddr.ToString()]["station_id"];
                                    packetInfo.stationLevel = stationsInfo[packetInfo.stationCode]["station_level"];
                                }
                                //20240221 包含IP，且device_type匹配，且只有一个设置，默认匹配
                                else if (dtmbMatchInfo.ContainsKey(addrKey) && dtmbMatchInfo[addrKey]["port"].Count == 1)
                                {
                                    foreach (string tmpPort in dtmbMatchInfo[addrKey]["port"].Keys)
                                    {
                                        if (dtmbMatchInfo[addrKey]["port"][tmpPort]["device_type"].Equals(packetInfo.header.device)
                                            && dtmbMatchInfo[addrKey]["port"][tmpPort]["station_id"].Equals(packetInfo.header.stationcode))
                                        {
                                            //得到台站信息
                                            ycdtmb = true;
                                            ycdtmbNo = dtmbMatchInfo[addrKey]["port"][tmpPort]["num"];
                                            //tunerDeviceType = dtmbMatchInfo[addrKey]["port"][tmpPort]["device_type"];
                                            ycdtmbDeviceType = dtmbMatchInfo[addrKey]["port"][tmpPort]["device_type"];
                                            packetInfo.tipMsg += "设备地址" + packetInfo.header.devaddr.ToString() + "未匹配" + dtmbMatchInfo[addrKey]["port"][tmpPort]["dtmb_dev"] +
                                                "且端口" + packetInfo.port + "未匹配" + tmpPort + "，系统默认通过解析，保留该提示";
                                            packetInfo.stationCode = dtmbMatchInfo[addrKey]["port"][tmpPort]["station_id"];
                                            packetInfo.stationLevel = stationsInfo[packetInfo.stationCode]["station_level"];
                                        }
                                    }
                                }

                                /******************************特殊处理************************************************/
                                //工位号是否设置在线标志
                                bool setOnline = false;
                                //IP是否是乱的，需要特殊处理
                                bool needSpecial = false;
                                /******************************特殊处理************************************************/

                                //判断台站ID是否为空来判断是否匹配
                                if (!"".Equals(packetInfo.stationCode))
                                {
                                    if (packetInfo.header != null)
                                    {
                                        //获取头信息
                                        //packetInfo.header = convertParse.GetHeaderData(packetInfo);
                                        //ErrorFlag是默认的，如果有异常，则不为no
                                        if (packetInfo.header != null && packetInfo.header.errorFlag.Equals("no"))
                                        {
                                            //心跳数据就不处理了
                                            if ("HEARTBEATMSG".Equals(packetInfo.header.msgType.ToUpper()))
                                            {
                                                lock (syncLock)
                                                {
                                                    HeartbeatNum++;
                                                }
                                                continue;
                                            }

                                            //切记此部分后续需要修正
                                            /****************20220224 根据和宋协商，暂时固定将村级台站3+1错误的设备编码0x0302修改成正确的0x0709*****************/
                                            if ("0X0302".Equals(packetInfo.header.device.ToUpper()) && "5".Equals(packetInfo.stationLevel)
                                                && (packetInfo.port == 2050 || packetInfo.port == 2060))
                                            {
                                                packetInfo.header.device = "0x0709";
                                            }
                                            /****************20220224 根据和宋协商，暂时固定将3+1错误的设备编码0x0302修改成正确的0x0709*****************/

                                            //同时将工位号也提前到这里赋值
                                            bool match = false;
                                            //六路调谐器特殊处理工位号，直接用tunerNo编号
                                            if (tuner)
                                            {
                                                if ("".Equals(packetInfo.deviceType) && !"".Equals(tunerDeviceType))
                                                {
                                                    packetInfo.deviceType = tunerDeviceType;
                                                }
                                                //工位号 台站编号+设备类型+工位编号
                                                packetInfo.positionId = packetInfo.stationCode + "_" + packetInfo.header.device + "_" + tunerNo;
                                                match = true;
                                            }
                                            //20230314 远场dtmb匹配
                                            else if (ycdtmb)
                                            {
                                                if ("".Equals(packetInfo.deviceType) && !"".Equals(ycdtmbDeviceType))
                                                {
                                                    packetInfo.deviceType = ycdtmbDeviceType;
                                                }
                                                //工位号 台站编号+设备类型+工位编号
                                                packetInfo.positionId = packetInfo.stationCode + "_" + packetInfo.header.device + "_" + ycdtmbNo;
                                                match = true;
                                            }

                                            //非六路调谐器
                                            //1~3级台站通过端口匹配
                                            if (!match && !needSpecial && ("1".Equals(packetInfo.stationLevel) || "2".Equals(packetInfo.stationLevel) || "3".Equals(packetInfo.stationLevel))
                                                && matchInfos["port"].ContainsKey(packetInfo.stationLevel)
                                                && matchInfos["port"][packetInfo.stationLevel].ContainsKey(packetInfo.port.ToString()))
                                            {
                                                if ("".Equals(packetInfo.deviceType) && matchInfos["port"][packetInfo.stationLevel][packetInfo.port.ToString()].ContainsKey("dev_type"))
                                                {
                                                    //packetInfo.deviceType = matchInfos["port"][packetInfo.stationLevel][packetInfo.port.ToString()]["dev_type"];
                                                    string tmpType = matchInfos["port"][packetInfo.stationLevel][packetInfo.port.ToString()]["dev_type"];
                                                    packetInfo.deviceType = tmpType.Split(',')[0];
                                                }
                                                //工位号 台站编号+设备类型+工位编号
                                                packetInfo.positionId = packetInfo.stationCode + "_" + packetInfo.header.device + "_" +
                                                    matchInfos["port"][packetInfo.stationLevel][packetInfo.port.ToString()]["work_no"];
                                                match = true;
                                            }
                                            //4、5级台站使用设备地址去匹配
                                            //else if (!match && !needSpecial && ("4".Equals(packetInfo.stationLevel) || "5".Equals(packetInfo.stationLevel))
                                            else if (!match && ("4".Equals(packetInfo.stationLevel) || "5".Equals(packetInfo.stationLevel))
                                                && matchInfos["devAddr"].ContainsKey(packetInfo.stationLevel)
                                                && matchInfos["devAddr"][packetInfo.stationLevel].ContainsKey(packetInfo.header.devaddr.ToString()))
                                            {
                                                if (matchInfos["devAddr"][packetInfo.stationLevel][packetInfo.header.devaddr.ToString()].ContainsKey("dev_type") &&
                                                    !"".Equals(matchInfos["devAddr"][packetInfo.stationLevel][packetInfo.header.devaddr.ToString()]["dev_type"]))
                                                {
                                                    string tmpType = matchInfos["devAddr"][packetInfo.stationLevel][packetInfo.header.devaddr.ToString()]["dev_type"];
                                                    //20220802 ysw 研发部设备通信协议4级的，主动读取的数据无设备类型，这里先通过IP和设备地址新增
                                                    if ("".Equals(packetInfo.deviceType))
                                                    {
                                                        packetInfo.tipMsg += ("".Equals(packetInfo.tipMsg) ? "" : ",") + "设备类型为空，通过设备地址【" + packetInfo.header.devaddr + "】更正为【" + tmpType.Split(',')[0] + "】";
                                                        packetInfo.deviceType = tmpType.Split(',')[0];
                                                        setOnline = true;
                                                    }
                                                    //20240812 4级不做修正
                                                    //判断设备类型是否正常 4、5级台站，根据设备地址，也可以反向找到设备类型
                                                    //else if (!tmpType.Equals(packetInfo.deviceType))
                                                    //else if (!tmpType.Contains(packetInfo.deviceType))
                                                    else if (!tmpType.Contains(packetInfo.deviceType) && "5".Equals(packetInfo.stationLevel))
                                                    {
                                                        packetInfo.tipMsg += ("".Equals(packetInfo.tipMsg) ? "" : ",") + "设备类型【" + packetInfo.deviceType +
                                                            "】异常，通过设备地址【" + packetInfo.header.devaddr + "】更正为【" + tmpType.Split(',')[0] + "】";
                                                        packetInfo.deviceType = tmpType.Split(',')[0];
                                                        setOnline = true;
                                                    }
                                                }

                                                //工位号 台站编号+设备类型+工位编号
                                                packetInfo.positionId = packetInfo.stationCode + "_" + packetInfo.header.device + "_" +
                                                    matchInfos["devAddr"][packetInfo.stationLevel][packetInfo.header.devaddr.ToString()]["work_no"];

                                                if (needSpecial)
                                                {
                                                    setOnline = true;
                                                }
                                                match = true;
                                            }
                                            //20240812新增 4、5级台站，也尝试通过端口匹配
                                            else if (!match && !needSpecial && ("4".Equals(packetInfo.stationLevel) || "5".Equals(packetInfo.stationLevel))
                                                && matchInfos["port"].ContainsKey(packetInfo.stationLevel)
                                                && matchInfos["port"][packetInfo.stationLevel].ContainsKey(packetInfo.port.ToString()))
                                            {
                                                if ("".Equals(packetInfo.deviceType) && matchInfos["port"][packetInfo.stationLevel][packetInfo.port.ToString()].ContainsKey("dev_type"))
                                                {
                                                    //packetInfo.deviceType = matchInfos["port"][packetInfo.stationLevel][packetInfo.port.ToString()]["dev_type"];
                                                    string tmpType = matchInfos["port"][packetInfo.stationLevel][packetInfo.port.ToString()]["dev_type"];
                                                    packetInfo.deviceType = tmpType.Split(',')[0];
                                                }
                                                //工位号 台站编号+设备类型+工位编号
                                                packetInfo.positionId = packetInfo.stationCode + "_" + packetInfo.header.device + "_" +
                                                    matchInfos["port"][packetInfo.stationLevel][packetInfo.port.ToString()]["work_no"];
                                                match = true;
                                            }

                                            /******************************特殊处理************************************************/
                                            //20221018 考虑到1~3级台站，有些设备地址匹配，端口无法匹配，这里也让通过解析，同时错误提示也保留
                                            //1~3级台站特殊处理，尝试通过设备地址匹配
                                            if (SpecialFlow && !match && ("1".Equals(packetInfo.stationLevel) || "2".Equals(packetInfo.stationLevel) || "3".Equals(packetInfo.stationLevel))
                                                && matchInfos["devAddr"].ContainsKey(packetInfo.stationLevel)
                                                && matchInfos["devAddr"][packetInfo.stationLevel].ContainsKey(packetInfo.header.devaddr.ToString()))
                                            {
                                                if ("".Equals(packetInfo.deviceType) && matchInfos["devAddr"][packetInfo.stationLevel][packetInfo.header.devaddr.ToString()].ContainsKey("dev_type"))
                                                {
                                                    //packetInfo.deviceType = matchInfos["devAddr"][packetInfo.stationLevel][packetInfo.header.devaddr.ToString()]["dev_type"];
                                                    string tmpType = matchInfos["devAddr"][packetInfo.stationLevel][packetInfo.header.devaddr.ToString()]["dev_type"];
                                                    packetInfo.deviceType = tmpType.Split(',')[0];
                                                }
                                                //工位号 台站编号+设备类型+工位编号
                                                packetInfo.positionId = packetInfo.stationCode + "_" + packetInfo.header.device + "_" +
                                                    matchInfos["devAddr"][packetInfo.stationLevel][packetInfo.header.devaddr.ToString()]["work_no"];

                                                packetInfo.tipMsg += ("".Equals(packetInfo.tipMsg) ? "" : ",") + "台站级别【" + packetInfo.stationLevel + "】或设备端口【" + packetInfo.port.ToString() +
                                                    "】不匹配异常，尝试通过设备地址【" + packetInfo.header.devaddr +
                                                    "】匹配处理，提示：1、2、3级台站匹配端口，4、5级台站匹配设备地址，此记录已通过解析，但保留该错误提示";

                                                setOnline = true;
                                                match = true;
                                            }
                                            //20221018 考虑到1~3级台站，有些设备地址匹配，端口无法匹配，这里也让通过解析，同时错误提示也保留

                                            //20221018 考虑到四级台站很多端口能匹配，设备地址无法匹配的，这里也让通过解析，同时错误提示也保留
                                            if (SpecialFlow && !match && "4".Equals(packetInfo.stationLevel)
                                                && matchInfos["port"].ContainsKey(packetInfo.stationLevel)
                                                && matchInfos["port"][packetInfo.stationLevel].ContainsKey(packetInfo.port.ToString()))
                                            {
                                                if ("".Equals(packetInfo.deviceType) && matchInfos["port"][packetInfo.stationLevel][packetInfo.port.ToString()].ContainsKey("dev_type"))
                                                {
                                                    //packetInfo.deviceType = matchInfos["port"][packetInfo.stationLevel][packetInfo.port.ToString()]["dev_type"];
                                                    string tmpType = matchInfos["port"][packetInfo.stationLevel][packetInfo.port.ToString()]["dev_type"];
                                                    packetInfo.deviceType = tmpType.Split(',')[0];
                                                }
                                                //工位号 台站编号+设备类型+工位编号
                                                packetInfo.positionId = packetInfo.stationCode + "_" + packetInfo.header.device + "_" +
                                                    matchInfos["port"][packetInfo.stationLevel][packetInfo.port.ToString()]["work_no"];

                                                packetInfo.tipMsg += ("".Equals(packetInfo.tipMsg) ? "" : ",") + "台站级别【" + packetInfo.stationLevel + "】或设备地址【" + packetInfo.header.devaddr +
                                                    "】不匹配异常，尝试通过设备端口【" + packetInfo.port.ToString() +
                                                    "】匹配处理，提示：1、2、3级台站匹配端口，4、5级台站匹配设备地址，此记录已通过解析，但保留该错误提示";

                                                setOnline = true;
                                                match = true;
                                            }
                                            //20221018 考虑到四级台站很多端口能匹配，设备地址无法匹配的，这里也让通过解析，同时错误提示也保留

                                            //20221018 村级台站存在设备地址、端口等都识别正常，但是配置设备类型错误，这里通过position_id_match匹配到设备类型，让解析通过，同时错误提示也保留
                                            if (SpecialFlow && !match && "5".Equals(packetInfo.stationLevel)
                                                && matchInfos["devAddr"].ContainsKey(packetInfo.stationLevel)
                                                && matchInfos["devAddr"][packetInfo.stationLevel].ContainsKey(packetInfo.header.devaddr.ToString()))
                                            {
                                                string tmpType = matchInfos["devAddr"][packetInfo.stationLevel][packetInfo.header.devaddr.ToString()]["dev_type"];
                                                if (!tmpType.Contains(packetInfo.deviceType))
                                                {
                                                    string tipMsg = "设备类型" + packetInfo.deviceType + "未未匹配，尝试通过position_id_match配置匹配到设备类型" + tmpType.Split(',')[0];
                                                    //position_id_match匹配到设备类型
                                                    tipMsg = tipMsg + tmpType.Split(',')[0];
                                                    tipMsg = tipMsg + "，此记录已通过解析，但保留该错误提示";

                                                    packetInfo.tipMsg += ("".Equals(packetInfo.tipMsg) ? "" : ",") + tipMsg;
                                                    packetInfo.deviceType = tmpType.Split(',')[0];

                                                    //工位号 台站编号+设备类型+工位编号
                                                    packetInfo.positionId = packetInfo.stationCode + "_" + packetInfo.header.device + "_" +
                                                        matchInfos["devAddr"][packetInfo.stationLevel][packetInfo.header.devaddr.ToString()]["work_no"];

                                                    setOnline = true;
                                                    match = true;
                                                }
                                            }
                                            //20221018 村级台站存在设备地址、端口等都识别正常，但是配置设备类型错误，这里通过position_id_match匹配到设备类型，让解析通过，同时错误提示也保留

                                            //20230129 如果还是无法确定工位号编号，尝试解析数据，根据频率尝试匹配
                                            //packetInfo.structData = convertParse.DataConvert(packetInfo);
                                            if (SpecialFlow && !match && positionIdToFrqs.ContainsKey(packetInfo.stationCode))
                                            {
                                                //尝试解析数据
                                                Dictionary<string, Object> structData = convertParse.DataConvert(packetInfo);
                                                if (structData != null)
                                                {
                                                    Dictionary<string, Object> tmpDatas = (Dictionary<string, Object>)structData["datas"];
                                                    if (tmpDatas != null)
                                                    {
                                                        packetInfo.structData = structData;
                                                        //台站频率匹配信息
                                                        Dictionary<string, Dictionary<string, string>> positionIdToFrq = positionIdToFrqs[packetInfo.stationCode];
                                                        //查找数据中的频率
                                                        List<string> dataFrq = tmpDatas.Keys.Where(k => k.ToUpper().Contains("FREQUENCY")).ToList();
                                                        if (dataFrq != null && dataFrq.Count > 0)
                                                        {
                                                            string frq = tmpDatas[dataFrq[0]].ToString();
                                                            List<string> locate = positionIdToFrq.Keys.Where(k => k.StartsWith(packetInfo.stationCode + "_" + packetInfo.deviceType)).ToList();
                                                            if (locate != null && locate.Count > 0)
                                                            {
                                                                foreach (string pid in locate)
                                                                {
                                                                    if (positionIdToFrq[pid].ContainsKey(frq))
                                                                    {
                                                                        string tipMsg = "通过positionIdToFrq历史缓存的频率数据匹配到工位号为" + pid + "，此记录已通过解析，但保留该错误提示";

                                                                        packetInfo.tipMsg += ("".Equals(packetInfo.tipMsg) ? "" : ",") + tipMsg;

                                                                        packetInfo.positionId = pid;
                                                                        setOnline = true;
                                                                        match = true;
                                                                        break;
                                                                    }
                                                                }
                                                            }
                                                        }
                                                    }
                                                }
                                            }

                                            /******************************特殊处理************************************************/

                                            if (packetInfo.positionId != null && !"".Equals(packetInfo.positionId))
                                            {
                                                //如果targetPositionId对应关系里面没有该IP:PORT，则需要写redis
                                                if (!targetPositionId.ContainsKey(from))
                                                {
                                                    setOnline = true;
                                                }
                                                else if (targetPositionId.Where(d => d.Value.Equals(packetInfo.positionId)).ToList().Count == 0)
                                                {
                                                    setOnline = true;
                                                }

                                                //172.18.0.1 172.17.0.1
                                                if ((packetInfo.tipMsg.Contains("172.") && packetInfo.tipMsg.Contains(".0.1")))
                                                {
                                                    setOnline = true;
                                                }

                                                //特殊处理IP无法识别的数据，这里需要写台站、设备在线标志
                                                if (setOnline)
                                                {
                                                    //这里也处理一下离线判断
                                                    redisCache.StringSetAsync("online_" + packetInfo.positionId, "0");
                                                    redisCache.StringSetAsync("newest_time_" + packetInfo.positionId, packetInfo.receivingTime);
                                                }

                                                //赋值设备序列号
                                                DeviceInfo deviceInfo = GetDeviceInfo(packetInfo);
                                                packetInfo.seviceSN = deviceInfo == null ? "" : deviceInfo.seviceSN;


                                                /**************新增解析内容*****************/
                                                if (packetInfo.structData == null)
                                                {
                                                    packetInfo.structData = ConvertParseImpls.ContainsKey(packetInfo.header.classVer) ? ConvertParseImpls[packetInfo.header.classVer].DataConvert(packetInfo) : null;
                                                }

                                                if (packetInfo.structData != null)
                                                {
                                                    //解析后的key_val
                                                    int paramId = int.Parse(((Dictionary<string, string>)packetInfo.structData["addr"])["paramId"]);
                                                    packetInfo.dictData = GetCols(paramId, (Dictionary<string, Object>)packetInfo.structData["datas"]);

                                                    packetInfo.cjDataAnaTime = DateTime.Now;
                                                    packetInfo.bodyNode = nodeName + "_" + (nodeNo + 1).ToString();
                                                    lock (dataCacheSet)
                                                    {
                                                        dataCacheSet.Add(packetInfo);
                                                    }

                                                    lock (syncLock)
                                                    {
                                                        gapNums++;
                                                    }

                                                    packError = false;
                                                }
                                                /**************新增解析内容*****************/
                                                //错误或者不需要解析的数据保存
                                                else
                                                {
                                                    if ("".Equals(packetInfo.tipMsg))
                                                    {
                                                        packetInfo.tipMsg = "帧体解析错误";
                                                    }
                                                }
                                            }
                                            //错误或者不需要解析的数据保存
                                            else
                                            {
                                                packetInfo.tipMsg += ("".Equals(packetInfo.tipMsg) ? "" : ",") + "台站级别【" + packetInfo.stationLevel + "】或设备地址【" + packetInfo.header.devaddr +
                                                    "】或者设备端口【" + packetInfo.port + "】不匹配异常，请检查position_id_match表（1、2、3级台站匹配端口，4、5级台站匹配设备地址）";
                                            }
                                        }
                                        else
                                        {
                                            if ("".Equals(packetInfo.tipMsg))
                                            {
                                                packetInfo.tipMsg = "帧头识别异常";
                                            }
                                        }
                                    }
                                    else
                                    {
                                        packetInfo.tipMsg = "未知帧体";
                                    }
                                }
                                else
                                {
                                    packetInfo.tipMsg = ("".Equals(packetInfo.tipMsg) ? "" : packetInfo.tipMsg + ",") + "IP未知";
                                    if (packetInfo.header != null && !"".Equals(packetInfo.header.stationcode) && stationsInfo.ContainsKey(packetInfo.header.stationcode))
                                    {
                                        packetInfo.tipMsg = ("".Equals(packetInfo.tipMsg) ? "" : packetInfo.tipMsg + ",") + "请核对，可能是台站【" + stationsInfo[packetInfo.header.stationcode]["station_name"] +
                                            "(" + stationsInfo[packetInfo.header.stationcode]["ip_addr"] + ")" + "】的数据";
                                        packetInfo.stationLevel = stationsInfo[packetInfo.header.stationcode]["station_level"];
                                    }

                                    if (packetInfo.header != null && packetInfo.structData == null)
                                    {
                                        packetInfo.structData = ConvertParseImpls.ContainsKey(packetInfo.header.classVer) ? ConvertParseImpls[packetInfo.header.classVer].DataConvert(packetInfo) : null;
                                    }
                                }
                            }

                            if (packError)
                            {
                                redisStatus.HashSetAsync("packetInfos_err", packetInfo.ip + ":" + packetInfo.port, packetInfo);
                            }
                        }
                    }
                }
                catch (Exception e)
                {
                    mLogger.LogError("解析数据异常，" + e);
                }
                finally
                {
                    Thread.Sleep(1);
                }
            }
        }

        /// <summary>
        /// 前置处理数据
        /// </summary>
        private void DealDataPre()
        {
            Thread.Sleep(4 * 1000);
            RedisHelper cacheRedis = new RedisHelper(redisConnString, mLogger);
            IBatch cacheBatch = cacheRedis.CreateBatch();
            while (true)
            {
                try
                {
                    if (dataCacheSet.Count > 0)
                    {
                        while (dataCacheSet.Count > 0)
                        {
                            PacketInfo packetInfo;
                            lock (dataCacheSet)
                            {
                                packetInfo = dataCacheSet.Take();
                            }
                            if (dataCacheSet.Count < 1000)
                            {
                                DataCache(packetInfo, cacheBatch, cacheRedis);
                                lock (syncLock)
                                {
                                    gapNums--;
                                }
                            }
                        }
                    }
                }
                catch (Exception e)
                {
                    mLogger.LogError("前置数据写缓存异常，" + e);
                }
                finally
                {
                    Thread.Sleep(1);
                }
            }
        }

        /// <summary>
        /// 数据处理
        /// 缓存服务无法处理时，在这里处理
        /// </summary>
        /// <param name="packet"></param>
        private void DataCache(PacketInfo packetInfo, IBatch batch, RedisHelper redis)
        {
            try
            {
                int paramId = -1;
                Dictionary<string, Object> data = packetInfo.structData;

                //判断最新时间是否大于redis保存的时间 大于则返回，否则会造成数据跳动
                //仅仅用于记录最新时间
                string redisKey = "newest_time_cache_" + packetInfo.positionId;
                DateTime newestSetTime = redis.StringGet<DateTime>(redisKey);
                if ((newestSetTime < packetInfo.receivingTime))
                {
                    redis.StringSetAsync(redisKey, packetInfo.receivingTime);
                }

                //数据是否不是最新
                //不是最新的数据，也写入，写入加入后缀PRE
                bool newdata = true;

                //新增msgType用于判断
                //redisKey = "newest_time_cache_" + packetInfo.positionId + "_" + packetInfo.header.msgType;
                redisKey = "newest_time_cache_" + packetInfo.positionId + "_" + packetInfo.header.msgType + "_" + packetInfo.header.bodysize;
                newestSetTime = redis.StringGet<DateTime>(redisKey);
                //if ((newestSetTime >= packetInfo.receivingTime))
                if ((newestSetTime >= packetInfo.receivingTime) && newestSetTime.Second != packetInfo.receivingTime.Second)
                {

                    if ("1".Equals(packetInfo.stationLevel) || "2".Equals(packetInfo.stationLevel) || "3".Equals(packetInfo.stationLevel))
                    {
                        if (newestSetTime == packetInfo.receivingTime)
                        {
                            mLogger.LogInformation(redisKey + "数据重复 时间" + " 节点=" + packetInfo.receiveNo + " " + packetInfo.pack_key);
                        }
                        else
                        {
                            mLogger.LogInformation(redisKey + "数据延迟 时间" + " 节点=" + packetInfo.receiveNo +
                                " 收到=" + TransForm.DateTimeFormat(packetInfo.receivingTime, "HH:mm:ss.fffffff") +
                                " 节点=" + packetInfo.receiveNode +
                                " 帧体=" + TransForm.DateTimeFormat(packetInfo.cjDataAnaTime, "HH:mm:ss.fffffff") +
                                " 当前最新=" + TransForm.DateTimeFormat(newestSetTime), "HH:mm:ss.fffffff");
                        }
                    }

                    if (!"0x0101".Equals(packetInfo.deviceType) && !"0x0706".Equals(packetInfo.deviceType))
                    {
                        return;
                    }

                    //不是最新的数据
                    newdata = false;
                }
                else
                {
                    redis.StringSetAsync(redisKey, packetInfo.receivingTime, TimeSpan.FromMinutes(10));
                }

                if (data.ContainsKey("addr"))
                {
                    //Dictionary<string, Object> addr = JsonConvert.DeserializeObject<Dictionary<string, Object>>(data["addr"].ToString());
                    Dictionary<string, string> addr = (Dictionary<string, string>)data["addr"];
                    paramId = addr.ContainsKey("paramId") ? int.Parse(addr["paramId"].ToString()) : paramId;
                }
                string schemaName = "";
                if (data.ContainsKey("ptlinfo"))
                {
                    //Dictionary<string, string> ptlinfo = JsonConvert.DeserializeObject<Dictionary<string, string>>(data["ptlinfo"].ToString());
                    Dictionary<string, string> ptlinfo = (Dictionary<string, string>)data["ptlinfo"];
                    schemaName = ptlinfo.ContainsKey("file_name") ? ptlinfo["file_name"].ToString() : schemaName;
                }
                if (paramId != -1)
                {
                    if (tableInfos.ContainsKey(paramId))
                    {
                        if (data.ContainsKey("datas"))
                        {
                            //批量写
                            List<HashEntry> hashFields = new();
                            //工位号数据key
                            string hashKey = "DATA_" + packetInfo.positionId;

                            if (newdata)
                            {
                                //最新时间也写入
                                //batch.MHashSetAsync(hashKey, "SCHEMANAME_" + GetMsgType(packetInfo), schemaName.ToLower().Substring(0, schemaName.LastIndexOf('.') == -1 ? 0 : schemaName.LastIndexOf('.')).Replace(".", "_"));
                                hashFields.Add(new HashEntry("SCHEMANAME_" + GetMsgType(packetInfo), schemaName.ToLower().Substring(0, schemaName.LastIndexOf('.') == -1 ? 0 : schemaName.LastIndexOf('.')).Replace(".", "_")));

                                //缓存时的系统时间
                                //hashFields.Add(new HashEntry("CACHE_TIME" + packetInfo.header.bodysize, TransForm.DateTimeFormat(DateTime.Now)));
                                hashFields.Add(new HashEntry("CACHE_TIME", TransForm.DateTimeFormat(DateTime.Now)));
                                //batch.MHashSetAsync(hashKey, "LATEST_TIME", TransForm.DateTimeFormat(packetInfo.receivingTime));
                                hashFields.Add(new HashEntry("LATEST_TIME", TransForm.DateTimeFormat(packetInfo.receivingTime)));
                                //hashFields.Add(new HashEntry("LATEST_TIME_" + packetInfo.header.bodysize, TransForm.DateTimeFormat(packetInfo.receivingTime)));

                                //加上厂家编号
                                //batch.MHashSetAsync(hashKey, "VENDOR", packetInfo.header.vendor);
                                hashFields.Add(new HashEntry("VENDOR", packetInfo.header.vendor));

                                //20230904 记录当前数据，从接收到处理结束，经过了哪些节点编号
                                //receiveNode接收节点，groupNode分组节点，bodyNode解析节点，cacheNode缓存节点
                                string msgType = packetInfo.header.msgType;
                                hashFields.Add(new HashEntry("receiveNode_" + msgType, packetInfo.receiveNode));
                                hashFields.Add(new HashEntry("groupNode_" + msgType, packetInfo.headNode));
                                hashFields.Add(new HashEntry("bodyNode_" + msgType, packetInfo.bodyNode));
                                hashFields.Add(new HashEntry("cacheNode_" + msgType, nodeName + "_" + (nodeNo + 1).ToString()));
                            }

                            //前缀的作用：有些配置数据，是根据频率或者主机号去读取的，如果不加前缀，相同前缀的数据返回，后面的会把前面的数据覆盖掉，所以加前缀以区分
                            //找到数据前缀
                            //所有数据
                            //Dictionary<string, string> allCols = GetCols(paramId, data["datas"].ToString());
                            Dictionary<string, string> allCols = packetInfo.dictData;
                            string prefixCols = tableInfos[paramId].tableCfg.prefixCols.ToUpper();
                            string prefixVal = "";
                            if (!"".Equals(prefixCols))
                            {
                                foreach (string prefixCol in prefixCols.Split(','))
                                {
                                    if (!"".Equals(prefixCol) && allCols.ContainsKey(prefixCol))
                                    {
                                        string val = allCols[prefixCol];
                                        if (prefixCol.ToUpper().Contains("FREQUENCY"))
                                        {
                                            if (TransForm.IsDecimalSign(val) || TransForm.IsInt(val))
                                            {
                                                prefixVal += "-" + val;
                                            }
                                            else
                                            {
                                                prefixVal += "-0";
                                                mLogger.LogError(packetInfo.positionId + "的" + packetInfo.header.msgType + "的返回" + prefixCol + "[" + val + "]的值异常");
                                            }
                                        }
                                        else if (prefixCol.ToUpper().Contains("HOSTNO"))
                                        {
                                            if (TransForm.IsInt(val) && int.Parse(val) < 30)
                                            {
                                                prefixVal += "-" + val;
                                            }
                                            else
                                            {
                                                prefixVal += "-0";
                                                mLogger.LogError(packetInfo.positionId + "的" + packetInfo.header.msgType + "的返回" + prefixCol + "[" + val + "]的值异常");
                                            }
                                        }
                                        else
                                        {
                                            prefixVal += "-" + val;
                                        }
                                    }
                                }
                            }

                            //循环所有数据
                            foreach (string key in allCols.Keys)
                            {
                                if (allCols[key] != null && !"".Equals(allCols[key]) && !"NULL".Equals(allCols[key]))
                                {
                                    //写redis
                                    if (!key.ToUpper().Equals("FUNCTIONCODE"))
                                    {
                                        //索引值，拼接上前缀
                                        //string indexKey = packetInfo.positionId + prefixVal + "_" + key.ToUpper();
                                        string indexKey = ("".Equals(prefixVal) ? "" : prefixVal.Substring(1) + "_") + key.ToUpper();
                                        string val = (allCols[key] == null || "NULL".Equals(allCols[key])) ? "" : allCols[key] ?? "";

                                        if (newdata)
                                        {
                                            //不是最新的数据，也写入，写入加入后缀PRE
                                            //batch.MHashSetAsync(hashKey, indexKey, val);
                                            hashFields.Add(new HashEntry(indexKey, val));
                                        }

                                        //此部分已经提前到转换的SetSourceSEQ方法单独提前处理
                                        //1~3级台站，信号源左右音柱信息序列处理
                                        //if (packetInfo.deviceType.Equals("0x0101") && key.ToUpper().Contains("TUNERCHNAMP"))
                                        //{
                                        //    batch.MHashSetAsync("SEQ_" + packetInfo.positionId, indexKey + "_" + packetInfo.receivingTime.Second.ToString(), val);
                                        //}
                                    }

                                    //特殊下发测试编码处理，目前只针对数字电视
                                    //if ("0x0706".Equals(packetInfo.deviceType) && key.ToUpper().Equals("N0X8800"))
                                    //{
                                    //    OperateModel operateModel = redis.StringGet<OperateModel>("oper_" + allCols[key].Trim());
                                    //    if (operateModel != null)
                                    //    {
                                    //        packetInfo.dataCacheTime = DateTime.Now;

                                    //        operateModel.rcv_time = packetInfo.receivingTime;
                                    //        operateModel.pack_key = packetInfo.pack_key;
                                    //        operateModel.packetInfo = packetInfo;

                                    //        //测试命令特殊处理，保存数据到redis，生命周期为30分钟
                                    //        redisInit.StringSet("oper_" + operateModel.oper_key, operateModel, TimeSpan.FromMinutes(30));

                                    //        //写task_detail_log表
                                    //        string sql = GetInsTaskLogSql(operateModel);
                                    //        string kafkakey = operateModel.deviceInfo.position_id;
                                    //        ThreadPool.QueueUserWorkItem(new WaitCallback(PublishSqlData), new Dictionary<string, string>() { { "dbtype", "1" }, { "key", kafkakey }, { "sql", sql } });
                                    //    }
                                    //}
                                }
                            }

                            //鉴于如六路调谐器（0x0901）设备的告警信息，发生消息的机制是有告警时发生，无告警时停止发生，所以这里统一对是告警上报的数据打一个特殊的标志
                            //模拟信号源软件控制器 0x0101
                            //1-3级台站电力软件控制器 0x0B01
                            //1-3级台站环境系统软件控制器 0x0E01
                            //以上都是和0x0901一样的处理
                            if (newdata && packetInfo.header.msgType.ToUpper().Contains("ALARM") && (
                                packetInfo.deviceType.Equals("0x0901") ||
                                packetInfo.deviceType.Equals("0x0101") ||
                                packetInfo.deviceType.Equals("0x0B01") ||
                                packetInfo.deviceType.Equals("0x0E01")
                                ))
                            {
                                //记录最新告警时间
                                //batch.MHashSetAsync(hashKey, "ALARM_TIME", TransForm.DateTimeFormat(packetInfo.receivingTime));
                                hashFields.Add(new HashEntry("ALARM_TIME", TransForm.DateTimeFormat(packetInfo.receivingTime)));

                                //告警参数字段，偶尔写
                                if (packetInfo.receivingTime.Minute % 10 == 0)
                                {
                                    //记录哪些字段是属于告警的，用于和状态等其他字段区别开来，后面告警服务会判断告警相对于状态的时间（_LATEST_TIME）是否超时，超时则告警字段返回空值，以消除告警
                                    //batch.MHashSetAsync(hashKey, "ALARM_PARAS", JsonConvert.SerializeObject(allCols.ToDictionary(d => d.Key.ToUpper(), d => "")));
                                    hashFields.Add(new HashEntry("ALARM_PARAS", JsonConvert.SerializeObject(allCols.ToDictionary(d => d.Key.ToUpper(), d => ""))));
                                }
                            }

                            batch.MHashSetAsync(hashKey, hashFields.ToArray());
                            //写一条packetInfo包的缓存，方便测试观察
                            batch.MHashSetAsync("packetInfos", packetInfo.positionId, JsonConvert.SerializeObject(packetInfo));
                            batch.Execute();
                        }
                        else
                        {
                            mLogger.LogError("参数datas缺失，请检查！");
                        }
                    }
                    else
                    {
                        mLogger.LogError("配置参数缺失，请检查数据库配置！");
                    }
                }
                else
                {
                    mLogger.LogError("参数addr-paramId缺失，请检查！");
                }
            }
            catch (Exception e)
            {
                mLogger.LogError("写redis缓存数据失败！ " + e);
            }
        }
        #endregion

        #region 写redis
        /// <summary>
        /// 定时清理垃圾端口
        /// 截断队列长度，避免数据无限增长
        /// </summary>
        private void SetTargetStatus()
        {
            Thread.Sleep(3000);
            IBatch batch = redisStatus.CreateBatch();
            while (true)
            {
                try
                {
                    //当前服务副本编号
                    int no = (nodeNo % nodeSum) + 1;
                    List<string> targets = targetRec.Keys.ToList();
                    foreach (string target in targets)
                    {
                        if ((DateTime.Now - targetRec[target]).TotalSeconds < 30)
                        {
                            //batch.MListTrimAsync("pack_" + target, 0, 50);
                            ////batch.MListTrimAsync("pack_" + target, 0, 10);
                            batch.MHashSetAsync("pack_addrs", target, TransForm.DateTimeFormat(targetRec[target]));
                            batch.MHashSetAsync("pack_matchs", target, nodeName + "_" + no.ToString());
                            //string pid = (targetPositionId.ContainsKey(target) ? targetPositionId[target] : "");
                            //batch.MHashSetAsync("pack_matchs", target, nodeName + "_" + no.ToString() + "," + pid);
                        }
                        else
                        {
                            lock (targetRec)
                            {
                                targetRec.Remove(target);
                            }
                        }
                    }
                    batch.Execute();
                }
                catch { }
                finally
                {
                    Thread.Sleep(5000);
                }
            }
        }
        #endregion

        #region 数据解析成key_val
        /// <summary>
        /// 从数据获取值
        /// </summary>
        /// <param name="paramId"></param>
        /// <param name="dataStr"></param>
        /// <param name="allCols"></param>
        private Dictionary<string, string> GetCols(int paramId, Dictionary<string, Object> dataStr)
        {
            Dictionary<string, string> allCols = new Dictionary<string, string>();
            try
            {
                Dictionary<string, Object> datas = dataStr;
                Dictionary<int, Object> dataPre = new Dictionary<int, object>();

                //是否结构体，非结构体即通用协议，通用协议时，value为空时不写入缓存
                bool isStruct = tableInfos[paramId].tableCfg.msgType.ToLower().Equals("struct");

                for (int i = 0; i <= tableInfos[paramId].layerNum; i++)
                {
                    dataPre.Add(i, null);
                }
                string col = "";
                string colTaos = "";
                //配置的字段 循环获取值
                bool updFlag = false;   //是否需要更新
                //动态字段判断处理
                string dynamicVal = "";
                foreach (int key in tableInfos[paramId].columns.Keys)
                {
                    int layer = tableInfos[paramId].columns[key].layer;
                    col = tableInfos[paramId].columns[key].columnName;
                    colTaos = tableInfos[paramId].columns[key].columnNameDb.ToUpper();
                    string type = tableInfos[paramId].columns[key].columnType;
                    string val = "";

                    //动态字段判断处理
                    dynamicVal = "";
                    updFlag = false;   //是否需要更新
                    if (tableInfos[paramId].dynamic.ContainsKey(col))
                    {
                        dynamicVal = tableInfos[paramId].dynamic[col];
                    }

                    //是末端节点
                    if (!tableInfos[paramId].columns[key].field.isNode)
                    {
                        if (layer == 0)
                        {
                            dataPre[layer] = datas.ContainsKey(col) ? datas[col] : "";
                            val = GetVal(type, dataPre[layer].ToString());

                            //dynamicVal非空（说明是动态字段）
                            //val是数字类型
                            //数值比较 dynamicVal小于val
                            if (!"".Equals(dynamicVal) && TransForm.IsInt(dataPre[layer].ToString()) && int.Parse(dynamicVal) < int.Parse(dataPre[layer].ToString()))
                            {
                                dynamicVal = dataPre[layer].ToString();
                                updFlag = true;
                                break;
                            }
                        }
                        else
                        {
                            if (dataPre[layer - 1] != null && !"".Equals(dataPre[layer - 1].ToString()))
                            {
                                try
                                {
                                    //数组
                                    List<Dictionary<string, Object>> dataIn = (List<Dictionary<string, Object>>)(dataPre[layer - 1]);
                                    string valIn = "";
                                    if (dataIn.Count > tableInfos[paramId].columns[key].loopIdx)
                                    {
                                        valIn = dataIn[tableInfos[paramId].columns[key].loopIdx].ContainsKey(col) ? dataIn[tableInfos[paramId].columns[key].loopIdx][col].ToString() : valIn;
                                    }
                                    val = GetVal(type, valIn);

                                    //dynamicVal非空（说明是动态字段）
                                    //val是数字类型
                                    //数值比较 dynamicVal小于val
                                    if (!"".Equals(dynamicVal) && TransForm.IsInt(valIn) && int.Parse(dynamicVal) < int.Parse(valIn))
                                    {
                                        dynamicVal = valIn;

                                        //需要更新
                                        updFlag = true;
                                        break;
                                    }
                                }
                                catch { }
                            }
                        }

                        if (isStruct)
                        {
                            allCols.TryAdd(colTaos, (val == null || "NULL".Equals(val)) ? "" : val);
                        }
                        //结构体时，只是写入有值的code
                        else if (val != null && !"".Equals(val) && !"NULL".Equals(val))
                        {
                            allCols.TryAdd(colTaos, val);
                        }
                    }
                    //父节点只负责给dataPre赋值
                    else
                    {
                        if (layer == 0)
                        {
                            dataPre[layer] = datas.ContainsKey(col) ? datas[col] : "";
                        }
                        else
                        {
                            if (!"".Equals(dataPre[layer - 1].ToString()))
                            {
                                try
                                {
                                    //数组
                                    List<Dictionary<string, Object>> dataIn = (List<Dictionary<string, Object>>)(dataPre[layer - 1]);
                                    if (dataIn.Count > tableInfos[paramId].columns[key].loopIdx)
                                    {
                                        //valIn = dataIn[tableInfos[paramId].columns[key].loopIdx].ContainsKey(col) ? dataIn[tableInfos[paramId].columns[key].loopIdx][col].ToString() : valIn;
                                        dataPre[layer] = dataIn[tableInfos[paramId].columns[key].loopIdx].ContainsKey(col) ? dataIn[tableInfos[paramId].columns[key].loopIdx][col] : null;
                                    }
                                    else
                                    {
                                        //dataPre[layer] = dataIn.Count > 0 ? dataIn[0] : null;
                                        dataPre[layer] = null;
                                    }
                                }
                                catch
                                {
                                    dataPre[layer] = "";
                                }
                            }
                            else
                            {
                                dataPre[layer] = "";
                            }
                        }
                    }
                }

                //需要更新
                if (updFlag)
                {
                    //更新表字段逻辑
                    //应该包含写表当前最大val的逻辑，否则下次还要执行一次
                    lock (tableInfos[paramId])
                    {
                        tableInfos[paramId].dynamic[col] = dynamicVal;
                        //tableInfos[paramId]重新初始化
                        tableInfos[paramId].Init();
                    }

                    //更新变量后，直接调用自身并返回
                    return GetCols(paramId, dataStr);
                }
            }
            catch { }

            return allCols;
        }

        /// <summary>
        /// 为了确保入库正确，这里要尝试做一些转换
        /// </summary>
        /// <param name="type"></param>
        /// <param name="val"></param>
        /// <returns></returns>
        private string GetVal(string type, string val)
        {
            string rs;
            val = val.Replace("\r", "").Replace(",", "，");
            val = val.Trim();
            if ("".Equals(val))
            {
                rs = "NULL";
            }
            else if ("FLOAT".Equals(type))
            {
                if (TransForm.IsDecimalSign(val))
                {
                    try
                    {
                        //val = float.Parse(val).ToString();
                        //新增为空判断
                        val = TransForm.IsNanInfinity(float.Parse(val)).ToString();
                    }
                    catch
                    {
                        val = "-1";
                    }
                    rs = val;
                }
                else
                {
                    rs = "NULL";
                }
            }
            else if ("INT".Equals(type))
            {
                if (TransForm.IsInt(val))
                {
                    try
                    {
                        int tmpI;
                        long tmpL;
                        if (int.TryParse(val, out tmpI))
                        {
                            val = tmpI.ToString();
                        }
                        else if (Int64.TryParse(val, out tmpL))
                        {
                            val = tmpL.ToString();
                        }
                        else
                        {
                            val = "NULL";
                        }
                    }
                    catch
                    {
                        val = "NULL";
                    }
                    rs = val;
                }
                else
                {
                    rs = "NULL";
                }
            }
            else if ("BIGINT".Equals(type))
            {
                if (TransForm.IsInt(val))
                {
                    try
                    {
                        val = long.Parse(val).ToString();
                    }
                    catch
                    {
                        val = "NULL";
                    }
                    rs = val;
                }
                else
                {
                    rs = "NULL";
                }
            }
            else
            {
                rs = val.Replace("'", "").Replace("", " ").Replace("  ", " ").Replace("\0", " ").Replace("\\", "").Replace(@"\r", "").Replace(@"\t", " ").Replace(@"\n", " ").Replace(@"\f", "").Replace(@"\b", "").Replace(@"\0", "");
            }
            return rs;
        }
        #endregion

        #region 命令下发
        ///// <summary>
        ///// 监听回调
        ///// </summary>
        ///// <param name="recieveResult"></param>
        //private void OnMessageRecieved(RecieveResult recieveResult)
        //{
        //    //回调时直接下发
        //    try
        //    {
        //        //堆积的不下发，避免出现批量多频次下发出现问题
        //        //启动后一定时间后才下发
        //        TimeSpan ts = DateTime.Now - dtBegin;
        //        //改成只用nodeName
        //        //string position_id_node = nodeName + "_" + (nodeNo + 1).ToString();
        //        string position_id_node = nodeName;fv
        //        if (ts.TotalSeconds > 90)
        //        //if (true)
        //        {
        //            OperateModel operateModel = JsonConvert.DeserializeObject<OperateModel>(recieveResult.Message);
        //            operateModel.sendkafka_time = recieveResult.Timestamp.UtcDateTime.AddHours(8);
        //            operateModel.send_node = nodeName + "_" + (nodeNo + 1).ToString();

        //            //改成只用nodeName
        //            int len = operateModel.deviceInfo.position_id_node.Contains('_') ? operateModel.deviceInfo.position_id_node.LastIndexOf('_') : 0;
        //            string tmpPidNode = operateModel.deviceInfo.position_id_node.Substring(0, len);
        //            //判断该命令是否应该由本节点下发
        //            bool send = false;
        //            //if (position_id_node.Equals(operateModel.deviceInfo.position_id_node))
        //            if (position_id_node.Equals(tmpPidNode))
        //            {
        //                send = true;
        //            }
        //            //当前是否第一个节点，即一号节点才该干的事情
        //            else if (nodeNo == 0)
        //            {
        //                if ("".Equals(tmpPidNode))
        //                {
        //                    send = true;
        //                }
        //                else
        //                {
        //                    //判断命令所带的position_id_node是否是当前运行的其他副本，如果不是，则默认由当前第一个节点发送
        //                    //if (redisInit.HashGet(keySvr, operateModel.deviceInfo.position_id_node) == null)
        //                    //if (redisInit.HashGet(keySvr, tmpPidNode) == null)
        //                    //改为用headKeys判断，不用redis
        //                    if (!headKeys.ContainsKey(tmpPidNode))
        //                    {
        //                        send = true;
        //                    }
        //                }
        //            }

        //            if (send)
        //            {
        //                byte[] cmd = TransForm.HexTobyte(operateModel.cmd, false);
        //                switch (operateModel.deviceInfo.ip_type)
        //                {
        //                    //ipv4
        //                    case "0":
        //                        mUdpNodeIpv4.Send(operateModel.deviceInfo.ip_addr, (ushort)operateModel.deviceInfo.port, cmd, cmd.Length);
        //                        break;
        //                    //ipv6
        //                    case "1":
        //                        mUdpNodeIpv6.Send(operateModel.deviceInfo.ip_addr, (ushort)operateModel.deviceInfo.port, cmd, cmd.Length);
        //                        break;
        //                    default: break;
        //                }
        //                if (!"QUICK".Equals(operateModel.logInfo) && !"SNMP".Equals(operateModel.logInfo))
        //                {
        //                    operateModel.send_time = DateTime.Now;
        //                    //把操作下发成功的也提交一份到Kafka，方便核对
        //                    ThreadPool.QueueUserWorkItem(new WaitCallback(PublishData), operateModel);

        //                    if (operateModel.oper_type != 0)
        //                    {
        //                        //测试命令特殊处理，保存数据到redis，生命周期为N分钟
        //                        //redisInit.StringSet("oper_" + operateModel.oper_key, operateModel, TimeSpan.FromMinutes(30));
        //                        redisInit.StringSet("oper_" + operateModel.oper_key, operateModel, TimeSpan.FromMinutes(5));
        //                    }
        //                    mLogger.LogInformation(nodeName + " 往" + operateModel.deviceInfo.ip_addr + ":" + operateModel.deviceInfo.port + " " + operateModel.deviceInfo.position_id + " 下发读取/控制命令 " + operateModel.logInfo + " 节点=" + (nodeNo + 1));
        //                }
        //            }
        //        }
        //    }
        //    catch (Exception e)
        //    {
        //        mLogger.LogError("命令下发失败，" + e.Message);
        //    }
        //    //接收后默认提交
        //    recieveResult.Commit();
        //}
        #endregion

        #region 写redis，写台站、设备在线信息
        /// <summary>
        /// 更新台站和设备在线信息数据
        /// </summary>
        private void UpdateOnlineStatus()
        {
            while (true)
            {
                try
                {
                    while (targetCache.Count > 0)
                    {
                        string pkgInfo;
                        IBatch batch = redisStatus.CreateBatch();
                        int cnt = 0;
                        while (targetCache.Count > 0)
                        {
                            lock (targetCache)
                            {
                                pkgInfo = targetCache.Take();
                            }

                            //太多来不及了，先丢弃
                            if (targetCache.Count > 3000)
                            {
                                continue;
                            }
                            string ip = pkgInfo.Split(":")[0];
                            string target = pkgInfo.Split(",")[0];
                            string rcvTime = pkgInfo.Split(",")[1].Replace("\"", "");
                            //台站数据是所有数据，适当丢弃一些也是可以的，所以有条件的丢弃
                            if (targetCache.Count < 1000)
                            {
                                if (targetStation.ContainsKey(ip))
                                {
                                    batch.MStringSetAsync("online_" + targetStation[ip], "0");
                                    batch.MStringSetAsync("newest_time_" + targetStation[ip], rcvTime);
                                }
                            }
                            //工位数据不能丢
                            if (targetPositionId.ContainsKey(target))
                            {
                                batch.MStringSetAsync("online_" + targetPositionId[target], "0");
                                batch.MStringSetAsync("newest_time_" + targetPositionId[target], rcvTime);
                            }

                            cnt++;

                            if (cnt > 10)
                            {
                                batch.Execute();
                                cnt = 0;
                            }
                        }
                        if (cnt > 0)
                        {
                            batch.Execute();
                        }
                    }
                }
                catch (Exception e)
                {
                    mLogger.LogError("更新台站和设备在线信息数据错误，" + e.ToString());
                }
                finally
                {
                    //Thread.Sleep(1);
                    TransForm.Sleep(1);
                }
            }
        }
        #endregion

        #region 辅助方法
        /// <summary>
        /// 获取数据包类型
        /// </summary>
        /// <param name="packetInfo"></param>
        /// <returns></returns>
        private string GetMsgType(PacketInfo packetInfo)
        {
            string msgType;
            string schemaName = packetInfo.header.msgType.ToUpper();
            if (schemaName.Contains("ALARM"))
            {
                msgType = "ALARM";
            }
            else if (schemaName.Contains("AUTO"))
            {
                msgType = "AUTO";
            }
            else if (schemaName.Contains("GET"))
            {
                msgType = "GET";
            }
            else
            {
                msgType = "BACK";
            }
            return msgType;
        }

        /// <summary>
        /// 获取转换实现类
        /// </summary>
        /// <param name="data">数据帧</param>
        /// <returns></returns>
        private IConvertParse MachingParse(byte[] data)
        {
            IConvertParse convertParse = null;

            //曾经报错过，添加判断，对于小于10的数据，可以判断为垃圾数据，不进行处理
            if (data.Length < 10)
            {
                return convertParse;
            }

            ////转为十六进制字符串
            string flag1 = TransForm.ByteToHexStringBigEndian(data.Skip(0).Take(1).ToArray()).ToUpper(); //第一字节
            string flag2 = TransForm.ByteToHexStringBigEndian(data.Skip(1).Take(1).ToArray()).ToUpper(); //第二字节

            //30W模拟调频发射机，50W自研数字电视发射机和水冷系统，第一字节0x25
            //add by ysw 20220802 30W模拟调频发射机, 50W自研数字电视发射机 主备切换器等 主动读取帧 即研发部设备通信协议的主动读取的数据 第一字节0x26 第二字节0x5A
            if (flag1.Equals("25") || (flag1.Equals("26") && flag2.Equals("5A")))
            {
                convertParse = convertParse_30Wor50W_Digit50W_Impl;
            }
            //300/1000W 数字电视发射机主动上报消息头
            //乡镇台站 50W
            //广西地面数字电视发射机通信接口协议
            else if (flag1.Equals("AA"))
            {
                //其实还有其它条件，第三字节为0x41，才是主动上报数据指令
                convertParse = convertParse_ByteParseToHeadDigitTransmitter300_Impl;
            }
            //远程监控通信协议规范（主控协议），命令标识码0x3c，远程监控通信协议规范的消息头[下发命令是为“>”，设备响应时为“<” 0x3c代表“<”]
            else if (flag1.Equals("3C"))
            {
                convertParse = convertParse_General_GXGD_Impl;
            }
            //四级环境采集器，第二字节：0x43。四级电源复位器，0x46
            //20221031 ysw 发现有第二字节0x91的数据，根据内容判断是四级电源复位器主动上报数据
            else if (flag2.Equals("43") || flag2.Equals("46") || flag2.Equals("91"))
            {
                convertParse = convertParse_Four_Enviroment_Power_NoHeader_Impl;
            }
            //snmp协议
            else if (flag1.Equals("30"))
            {
                convertParse = convertParse_SNMP_Impl;
            }

            return convertParse;
        }


        /// <summary>
        /// 获取设备信息
        /// </summary>
        /// <param name="packetInfo"></param>
        /// <returns></returns>
        private DeviceInfo GetDeviceInfo(PacketInfo packetInfo)
        {
            //判断并记录设备信息
            //得到设备信息，是为了获取seviceSN设备序列号
            DeviceInfo deviceInfo;

            try
            {
                //先从缓存判断
                if (deviceInfos.ContainsKey(packetInfo.positionId))
                {
                    deviceInfo = deviceInfos[packetInfo.positionId];
                    deviceInfo.new_time = DateTime.Now;

                    //姑且认为端口、台站级别和台站类型有可能会变化吧，更新一下到redis
                    //position_id_node 本工位号从哪个采集节点接收数据的，采集服务重启时会改变
                    if (!deviceInfo.port.ToString().Equals(packetInfo.port.ToString()) ||
                        !deviceInfo.ip_addr.Equals(packetInfo.ip) ||
                        !deviceInfo.staionLevel.Equals(packetInfo.stationLevel) ||
                        !deviceInfo.position_id_node.Equals(packetInfo.receiveNode) ||
                        !deviceInfo.dev_addr.Equals(packetInfo.header.devaddr.ToString())
                        )
                    {
                        deviceInfo.port = packetInfo.port;
                        deviceInfo.ip_addr = packetInfo.ip;
                        deviceInfo.staionLevel = packetInfo.stationLevel;
                        deviceInfo.position_id_node = packetInfo.receiveNode;
                        deviceInfo.dev_addr = packetInfo.header.devaddr.ToString();
                        redisCache.HashSet("station_deviceInfos", packetInfo.positionId, deviceInfo);
                    }
                }
                else
                {
                    //再从redis获取
                    deviceInfo = redisCache.HashGet<DeviceInfo>("station_deviceInfos", packetInfo.positionId);
                    if (deviceInfo == null)
                    {
                        //redis也没有，则新增要给，然后写入redis和记录到缓存
                        deviceInfo = new DeviceInfo
                        {
                            seviceSN = System.Guid.NewGuid().ToString("N"),
                            stationCode = packetInfo.stationCode,
                            staionLevel = packetInfo.stationLevel,
                            position_id = packetInfo.positionId,
                            position_id_node = packetInfo.receiveNode,
                            deviceType = packetInfo.header.device,
                            ip_addr = packetInfo.ip,
                            ip_type = TransForm.IsIPV6(packetInfo.ip) ? "1" : "0",
                            port = packetInfo.port,
                            dev_addr = packetInfo.header.devaddr.ToString()
                        };

                        deviceInfo.new_time = DateTime.Now;
                        redisCache.HashSet("station_deviceInfos", packetInfo.positionId, deviceInfo);
                    }
                    else
                    {
                        deviceInfo.new_time = DateTime.Now;
                    }

                    lock (syncLock)
                    {
                        //从redis获取后，放入缓存
                        deviceInfos.TryAdd(packetInfo.positionId, deviceInfo);
                    }
                }
            }
            catch
            {
                deviceInfo = null;
            }

            return deviceInfo;
        }
        #endregion

        #region 日志打印
        /// <summary>
        /// 定时监视判断是否增加线程池数量来处理数据
        /// </summary>
        private void TimerTask()
        {
            long rate;    //数量差
            long numsRcvPre = 0;    //上次数量
            int times = 0;
            while (true)
            {
                try
                {
                    rate = numsRcv - numsRcvPre;
                    numsRcvPre = numsRcv;

                    //以十秒钟为单位，统计接收频率
                    nodeStatusInfo["GetRate"] = rate;

                    //数据有变化，打印日志
                    if (rate > 0 || gapNums > 0 || skipNum > 0 || dataCacheGet.Count > 0 || targetCache.Count > 0 )
                    {
                        //打印服务器名称
                        mLogger.LogInformation(nodeName +
                            //" 线程池=" + maxPoolNum +
                            " 线程池=" + ThreadPool.ThreadCount +
                            " 设备数量=" + targetRec.Count +
                            " 数据帧=" + numsRcv.ToString() + " 待提交=" + dataCacheGet.Count +
                            " 接收帧频率=" + rate + " 提交缓存帧=" + gapNums + " 丢弃=" + skipNum + " redis写缓存=" + targetCache.Count +
                            " 节点数=" + nodeNum + " 当前节点号=" + (nodeNo + 1));
                    }
                    else
                    {
                        mLogger.LogInformation(nodeName + " 无数据 节点数 = " + nodeNum + " 当前节点号 = " + (nodeNo + 1));
                    }

                    lock (skipLock)
                    {
                        skipNum = 0;
                    }

                    //十分钟更新一次缓存
                    if (times % 6 == 0)
                    {
                        InitCacheInfo();
                    }

                    times++;
                    if (times > 10 * 10 * 10 * 10000)
                    {
                        times = 1;
                    }

                    //过期的key
                    List<string> tmpExpireKeys = rcvDataRecords.Where(d => d.Value.Count == 0 || (d.Value.Count > 0 && (DateTime.Now - d.Value.Last()).TotalMinutes > 5)).ToDictionary(d => d.Key, d => d.Value).Keys.ToList();
                    lock (rcvDataRecords)
                    {
                        foreach (string expireKey in tmpExpireKeys)
                        {
                            rcvDataRecords.Remove(expireKey);
                        }
                    }

                    if (rcvErrDevs.Count > 0)
                    {
                        mLogger.LogInformation("有数据异常上报设备：" + JsonConvert.SerializeObject(rcvErrDevs));
                        lock (rcvErrDevs)
                        {
                            rcvErrDevs.Clear();
                        }
                    }
                }
                catch (Exception e)
                {
                    mLogger.LogError("日志打印错误，" + e.ToString());
                }
                finally
                {
                    Thread.Sleep(10 * 1000);

                    //超过10分钟无数据，则让系统自己报错，在服务中，当前服务就会关闭，然后系统会自己重新启动
                    int m = 2;
                    //if ((DateTime.Now - dtNewestTime).TotalMinutes > 10)
                    if ((DateTime.Now - dtNewestTime).TotalMinutes > m)
                    {
                        mLogger.LogInformation("服务超过" + m + "分钟（最近" + TransForm.DateTimeFormat(dtNewestTime) + "）接收不到数据，系统主动抛出异常,本服务将关闭，新服务将启动！");
                        //转换错误抛出错误
                        int.Parse("a");
                    }
                }
            }
        }
        #endregion

        #region 心跳服务
        /// <summary>
        /// 心跳服务，定时往redis发送心跳数据
        /// 用于获取当前服务副本数量
        /// </summary>
        private void HeartBeatTask()
        {
            string lockKey = "lockKeyDataReceiveService";   //心跳节点锁key，保证唯一
            string uuid = nodeName;
            int gapSec = 2; //循环判断间隔
                            //int gapCheck = 11;   //判断时间差间隔，适当大于gapSec，如果大于，则判断为该节点离线，移除掉
            int gapCheck = 20;   //判断时间差间隔，适当大于gapSec，如果大于，则判断为该节点离线，移除掉

            //每次启动清空
            Thread.Sleep(new Random().Next(1, (gapSec + 1)) * 1000);

            //开始先写一条服务启动时间，用于给离线判断服务判断启动时间2分钟内不做离线判断，避免重启采集服务造成台站上报超时的告警
            //TimeSpan.FromMinutes(2) key存在分钟数
            redisInit.StringSet("DataReceiveService_start", "", TimeSpan.FromMinutes(2));

            while (true)
            {
                try
                {
                    DateTime dtNow = DateTime.Now;

                    //更新自身节点最新时间
                    redisInit.HashSet(keySvr, uuid, dtNow);

                    //记录统计信息，也作为心跳数据记录的一部分
                    nodeStatusInfo["Datetime"] = TransForm.DateTimeFormat(dtNow);
                    nodeStatusInfo["GetSum"] = numsRcv;
                    nodeStatusInfo["RedisWaiting"] = dataCacheGet.Count;
                    nodeStatusInfo["RedisCache"] = gapNums;
                    nodeStatusInfo["SkipNum"] = skipNum;
                    nodeStatusInfo["nodeNo"] = nodeNo + 1;
                    redisInit.HashSet("STATUS_" + keySvr, uuid, nodeStatusInfo);

                    //判断副本数最新时间 锁过期时间为5秒
                    Object obj = redisInit.LockAndDo(lockKey, uuid, TimeSpan.FromSeconds(gapSec),
                        () =>
                        {
                            //获取key下所有节点及其时间
                            Dictionary<string, DateTime> heartInfo = redisInit.HashGetAllDict<DateTime>(keySvr);
                            headKeys = heartInfo;
                            if (heartInfo != null && heartInfo.Count > 0)
                            {
                                //判断节点是否超时gapCheck，超时则移除
                                //判断当前节点ID再心跳节点的位置
                                int idx = 0;
                                foreach (string node in heartInfo.Keys)
                                {
                                    if (uuid.Equals(node))
                                    {
                                        //节点ID位置
                                        nodeNo = idx;
                                    }

                                    //判断节点过期时间
                                    //if ((DateTime.Now - heartInfo[node]).TotalSeconds > gapCheck)
                                    if ((dtNow - heartInfo[node]).TotalSeconds > gapCheck)
                                    {
                                        redisInit.HashDelete(keySvr, node);
                                        redisInit.HashDelete("STATUS_" + keySvr, node);
                                    }
                                    else
                                    {
                                        idx++;
                                    }
                                }

                                Dictionary<string, string> statusInfo = redisInit.HashGetAllDict<string>("STATUS_" + keySvr);
                                if (statusInfo != null && statusInfo.Count > 0 && statusInfo.Count != heartInfo.Count)
                                {
                                    //状态过期数据删除
                                    foreach (string node in statusInfo.Keys)
                                    {
                                        if (!heartInfo.ContainsKey(node))
                                        {
                                            redisInit.HashDelete("STATUS_" + keySvr, node);
                                        }
                                    }
                                }

                                //获得节点数
                                nodeNum = idx;
                            }
                            return true;
                        });

                    //锁异常
                    if (obj is null)
                    {
                        redisInit.HashSet(keySvr, uuid, dtNow);
                    }
                }
                finally
                {
                    Thread.Sleep(gapSec * 1000);
                }
            }
        }
        #endregion

        /// <summary>
        /// 统计数据写入redis
        /// </summary>
        private void Statistics()
        {
            Stopwatch sw;
            //判断间隔
            int interval = 1000;
            long numsRcvPre = numsRcv;    //上次数量
            string keyPre = "";

            Thread.Sleep(interval);
            while (true)
            {
                sw = Stopwatch.StartNew();
                try
                {
                    long tmpNum = numsRcv;
                    long secondCount = tmpNum - numsRcvPre;
                    numsRcvPre = tmpNum;

                    DateTime dt = DateTime.Now;

                    //记录最近两年的数据
                    //以哈希存储，以天为key，小时分钟为子subKey，总计365个key，每个key有24*60=1440个subKey
                    string key = "STAT_" + TransForm.DateTimeFormat(dt, "yyyyMMdd");
                    string subKey = TransForm.DateTimeFormat(dt, "HHmm");
                    Task<long> task = redisInit.HashIncrementAsync(key, subKey, secondCount);

                    //设置过期时间
                    if (!key.Equals(keyPre))
                    {
                        redisInit.KeyExpire(key, TimeSpan.FromDays(365 * 2));
                    }
                    keyPre = key;
                }
                catch { }
                finally
                {
                    int gap = (int)sw.ElapsedMilliseconds;
                    Thread.Sleep(gap >= interval ? 0 : (interval - gap));
                }
            }
        }

        /// <summary>
        /// 启动任务
        /// </summary>
        /// <param name="stoppingToken"></param>
        /// <returns></returns>
        protected override Task ExecuteAsync(CancellationToken stoppingToken)
        {
            mLogger.LogInformation("数据接收服务已启动");

            UdpNodeInit();  //开启UdpNode监听

            new Thread(HeartBeatTask).Start();      //心跳
            new Thread(Statistics).Start();         //统计
            new Thread(SetTargetStatus).Start();    //实时更新记录IP:PORT的最新时间
            new Thread(DealDataPre).Start();        //数据写缓存
            new Thread(DealData).Start();           //缓存数据处理线程
            new Thread(UpdateOnlineStatus).Start(); //更新台站和设备在线信息数据
            new Thread(TimerTask).Start();          //定时监视判断是否增加线程池数量来处理数据

            return Task.CompletedTask;
        }
    }
}
